package com.zhisheng.sql.blink.stream.example;

import com.zhisheng.sql.blink.stream.example.util.EmailSender;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

public class PlayerBalanceMonitorJob {
    public static void main(String[] args) throws Exception {
        // 设置执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        EnvironmentSettings settings = EnvironmentSettings.newInstance().inStreamingMode().build();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, settings);

        // 创建源表 - 假设有一个包含玩家投注信息的Kafka主题
        tableEnv.executeSql(
                "CREATE TABLE t_player_bet ("
                        + "  bet_id INT,"
                        + "  player_id INT,"
                        + "  player_email STRING,"
                        + "  bet_amount DOUBLE,"
                        + "  win_amount DOUBLE,"
                        + "  bet_time BIGINT,"
                        + "  ts AS TO_TIMESTAMP_LTZ(bet_time, 3),"
                        + "  WATERMARK FOR ts AS ts - INTERVAL '5' SECOND"
                        + ") WITH ("
                        + "  'connector' = 'kafka',"
                        + "  'topic' = 'mysql-server.test_sm.t_player_bet',"
                        + "  'properties.bootstrap.servers' = '192.168.50.235:9092,192.168.50.235:9093,192.168.50.235:9094',"
                        + "  'properties.group.id' = 'player_balance_monitor',"
                        + "  'scan.startup.mode' = 'latest-offset',"
                        + "  'format' = 'debezium-json',"
                        + "  'debezium-json.schema-include' = 'true'"
                        + ")");

        // SQL查询：计算每个玩家的总余额
        String playerBalanceQuery = 
                "SELECT "
                + "    player_id, "
                + "    player_email, "
                + "    SUM(win_amount - bet_amount) as total_balance "
                + "FROM t_player_bet "
                + "GROUP BY player_id, player_email "
                + "HAVING SUM(win_amount - bet_amount) >= 1000";

        Table resultTable = tableEnv.sqlQuery(playerBalanceQuery);

        // 转换为DataStream并添加告警处理
        DataStream<Row> resultStream = tableEnv.toChangelogStream(resultTable);

        // 配置邮件发送器
        EmailSender emailSender = new EmailSender(
                "your-email@example.com",  // 替换为实际的邮箱账号
                "your-password",           // 替换为实际的邮箱密码
                "smtp.example.com",        // 替换为实际的SMTP服务器
                587                        // 替换为实际的SMTP端口
        );

        // 添加邮件发送Sink
        resultStream.addSink(new SinkFunction<Row>() {
            @Override
            public void invoke(Row row, Context context) {
                try {
                    int playerId = (int) row.getField(0);
                    String playerEmail = (String) row.getField(1);
                    double balance = (double) row.getField(2);

                    String subject = "Balance Alert - Player " + playerId;
                    String content = String.format(
                            "Dear Player,\n\n" +
                            "Congratulations! Your current balance has reached %.2f.\n\n" +
                            "Best regards,\n" +
                            "Casino Team", 
                            balance);

                    emailSender.sendEmail(playerEmail, subject, content);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        });

        // 执行作业
        env.execute("Player Balance Monitor Job");
    }
}
